import beans.OrderEvent;
import beans.OrderResult;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author zkq
 * @date 2022/10/15 17:05
 */
public class OrderTimeoutWithoutCEP {
    //测输出流标签
    private final  static OutputTag<OrderResult> orderTimeoutTag = new OutputTag<OrderResult>("order-timeout"){};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<OrderEvent> orderStream = env.readTextFile("F:\\javasecode220620\\UserBehaviorAnalysis\\OrderTimeoutDetect\\src\\main\\resources\\OrderLog.csv")
                .map(line -> {
                    String[] split = line.split(",");
                    return new OrderEvent(new Long(split[0]), split[1], split[2], new Long(split[3]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        })
                );

        //自定义处理函数，主流输出正常支付事件，测输出流输出超时报警事件
        SingleOutputStreamOperator<OrderResult> resultStream = orderStream
                .keyBy(OrderEvent::getOrderId)
                .process(new OrderPayMatchDetect());

        resultStream.print("payed normally");
        resultStream.getSideOutput(orderTimeoutTag).print("timeout");

        env.execute("order timeout detect without cep");
    }

    public static class OrderPayMatchDetect extends KeyedProcessFunction<Long,OrderEvent,OrderResult>{
        //定义状态，保存之前订单是否已经来过 create，pay事件
        private ValueState<Boolean> isPayState;
        private ValueState<Boolean> isCreateState;
        //定义状态，保存定时器时间戳
        private ValueState<Long> timerTsState;

        @Override
        public void open(Configuration parameters) throws Exception {
            isCreateState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-created",Boolean.class,false));
            isPayState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-payed",Boolean.class,false));
            timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts",Long.class));
        }

        @Override
        public void processElement(OrderEvent value, Context ctx, Collector<OrderResult> out) throws Exception {
            //先获取当前状态
            Boolean isPayed = isPayState.value();
            Boolean isCreate = isCreateState.value();
            Long timerTs = timerTsState.value();

            //判断当前事件类型
            if ("create".equals(value.getEventType())){
                //1.如果来的是create，要判断是否支付过
                if (isPayed){
                    //1.1如果已经支付，输出正常匹配结果
                    out.collect(new OrderResult(value.getOrderId(),"payed successfully"));
                    //清空状态，删除定时器
                    isCreateState.clear();
                    isPayState.clear();
                    timerTsState.clear();
                    ctx.timerService().deleteEventTimeTimer(timerTs);
                }else {
                    //1.2如果没有支付过，注册15分钟后的定时器，开始等待支付事件
                    Long ts = (value.getTimestamp() + 15 * 60) * 1000L;
                    ctx.timerService().registerEventTimeTimer(ts);
                    timerTsState.update(ts);
                    isCreateState.update(true);
                }
            }else if ("pay".equals(value.getEventType())){
                //2.如果来的是pay，要判断是否有下单事件来过
                if (isCreate){
                    //2.1已经有过下单事件,要继续判断支付的时间戳是否超过15分钟
                    //虽然已经注册过定时器，但是可能乱序比较大 水位线还没到 就来了超时支付的数据
                    if (value.getTimestamp()*1000 < timerTs){
                        //2.1.1 在15分钟内未超时，正常匹配输出
                        out.collect(new OrderResult(value.getOrderId(),"payed successfully"));
                    }else {
                        //2.1.2 已经超时，输出测输出流报警
                        ctx.output(orderTimeoutTag,new OrderResult(value.getOrderId(),"payed but already timeout"));
                    }
                    //统一清空状态
                    isCreateState.clear();
                    isPayState.clear();
                    timerTsState.clear();
                    ctx.timerService().deleteEventTimeTimer(timerTs);
                }else {
                    //2.2 没有下单事件，乱序，注册一个定时器等待下单事件到来
                    ctx.timerService().registerEventTimeTimer(value.getTimestamp() * 1000L);
                    timerTsState.update(value.getTimestamp()*1000L);
                    isPayState.update(true);
                }
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception {
            //定时器触发，说明一定有一个事件没来，判断是哪个
            if ((isPayState.value())){
                //如果pay来了，说明create没来
                ctx.output(orderTimeoutTag,new OrderResult(ctx.getCurrentKey(),"keyed but not found created"));
            }else {
                //如果pay没来，支付超时
                ctx.output(orderTimeoutTag,new OrderResult(ctx.getCurrentKey(),"timeout"));
            }
            //清空状态
            isCreateState.clear();
            isPayState.clear();
            timerTsState.clear();
        }
    }

}
